package com.example.mq.mqserver.datacenter;

import com.example.mq.common.Consumer;
import com.example.mq.common.ConsumerEnv;
import com.example.mq.common.MqException;
import com.example.mq.mqserver.VirtualHost;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.util.concurrent.*;

/**
 * 通过这个类，来实现消费消息的核心逻辑
 */
public class ConsumerManager {
    // 持有上层的 VirtualHost 对象的引用，用来操作数据
    private VirtualHost parent;

    // 指定一个线程池，执行具体的回调任务
    private ExecutorService workPool = Executors.newFixedThreadPool(4);
    // 存放一个 令牌（queueName）的队列
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 扫描线程
    private Thread scanThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;
        scanThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 从阻塞队列中 拿到 队列名字
                    String queueName = tokenQueue.take();
                    // 2. 根据队列名字找到队列
                    MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 取令牌后发现，队列名不存在！queueName=" + queueName);
                    }
                    // 3. 从队列中消费消息
                    synchronized (queue) {
                        consumeMessage(queue);
                    }
                } catch (InterruptedException | MqException e) {
                    e.printStackTrace();
                }
            }
        });
        // 把线程设为后台线程
        scanThread.setDaemon(true);
        scanThread.start();
    }


    //
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    // 记录当前队列有哪些消费这订阅了
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在！queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有消息了，需要立即消费掉
            int messageCount = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < messageCount; i++) {
                // 这个方法调用一次就消费一条消息
                consumeMessage(queue);
            }
        }
    }

    private void consumeMessage(MSGQueue queue) {
        // 1. 先按照轮询的方式，找个消费者出来
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 当前队列没有消费者，暂时不消费
            return;
        }
        // 2. 从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            // 当前队列中没有消息，就不消费
            return;
        }
        // 3. 把消息带入到消费者的回调函数中，丢给线程池执行
        workPool.submit(() -> {
            try {
                // 1. 把消息放入到待确认集合中, 这个操作在执行回调之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 真正执行回调
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumeTag(), message.getBasicProperties(), message.getBody());
                // 3. 如果当前是自动应答，此时就可以消息删除
                // 手动应答，先不处理，交给后续消费者调用 basicAck来处理
                if (luckyDog.isAutoAck()) {
                    // 1) 删除硬盘
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除待确认集合
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中消息中心
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费！queueName=" + queue.getName());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }
}
